package app;

import com.alibaba.fastjson.JSON;
import domain.Access;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.summarize.aggregation.ValueSummaryAggregator;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.shaded.guava18.com.google.common.hash.BloomFilter;
import org.apache.flink.shaded.guava18.com.google.common.hash.Funnels;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * Create by zhh on 2022/9/9.
 * 新老用户的统计分析
 * 根据每个设备的device来判断是否是新老用户
 * 状态+布隆过滤器
 */
public class OsUserAppV3 {
    public static void main(String[] args) throws Exception{
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        SingleOutputStreamOperator<Access> cleanData = env.readTextFile("data/access.json").map(new MapFunction<String, Access>() {
            @Override
            public Access map(String s) throws Exception {
                //TODO... json==>Access

                try {

                    return JSON.parseObject(s, Access.class);
                } catch (Exception e) {
                    e.printStackTrace();
                    return null;
                }


            }
        })
                .filter(x -> x != null)
                .filter(new FilterFunction<Access>() {
                    @Override
                    //过滤启动的用户
                    public boolean filter(Access access) throws Exception {
                        return "startup".equals(access.event);
                    }
                });

        cleanData.keyBy(x->x.deviceType)//根据设备类型keyby
                .process(new KeyedProcessFunction<String, Access, Access>() {
                    private transient ValueState<BloomFilter<String>> state;
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        /**
                         * * @param name The (unique) name for the state.
                         * * @param typeInfo The type of the values in the state.
                         */
                        //使用布隆过滤器的状态描述,来一个设备，就放到布隆过滤器中
                        ValueStateDescriptor<BloomFilter<String>> descriptor=new ValueStateDescriptor<>("s", TypeInformation.of(new TypeHint<BloomFilter<String>>() {}));
                        state = getRuntimeContext().getState(descriptor);
                    }
                    //来一个数据，处理一个
                    @Override
                    public void processElement(Access value, Context ctx, Collector<Access> out) throws Exception {
                        String device = value.device;
                        BloomFilter<String> bloomFilter = state.value();
                        //如果布隆过滤器为空 初始化
                        if (null==bloomFilter){
                            bloomFilter = BloomFilter.create(Funnels.unencodedCharsFunnel(),100000);
                        }
                        //mightContain可能包含
                        if (!bloomFilter.mightContain(device)){//肯定不存在
                            bloomFilter.put(device);
                            value.nu = 1;//标记为新用户
                            state.update(bloomFilter);
                        }
                        out.collect(value);
                    }
                }).print();

        env.execute("OsUserAppV3");
    }

}


